package gossip.v1;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import static gossip.v1.GossipConfig.INDIRECT_CHECK_COUNT;
import static gossip.v1.GossipConfig.SUSPECT_THRESHOLD;

public class FailureDetector {
    private final GossipCore gossipCore;
    private final Map<String, Long> lastHeardTimes = new ConcurrentHashMap<>();
    private final Map<String, AtomicInteger> suspicionLevels = new ConcurrentHashMap<>();

    public FailureDetector(GossipCore gossipCore) {
        this.gossipCore = gossipCore;
    }

    public void check() {
        long now = System.currentTimeMillis();
        Map<String, MemberState> members = gossipCore.getMembershipTable().getMembers();

        // 更新本地节点时间戳
        gossipCore.getMembershipTable().getLocalState().updateTimestamp();

        for (MemberState member : members.values()) {
            if (member == gossipCore.getMembershipTable().getLocalState()) continue;

            Long lastHeard = lastHeardTimes.get(member.getNodeId());
            if (lastHeard == null) {
                lastHeardTimes.put(member.getNodeId(), now);
                suspicionLevels.put(member.getNodeId(), new AtomicInteger(0));
            } else if (now - lastHeard > SUSPECT_THRESHOLD) {
                handleSuspectNode(member.getNodeId());
            } else {
                // 重置怀疑级别
                suspicionLevels.get(member.getNodeId()).set(0);
            }
        }
    }

    public void recordActivity(SocketAddress address) {
        if (address instanceof InetSocketAddress) {
            InetSocketAddress inetAddr = (InetSocketAddress) address;
            // 使用与GossipCore一致的节点ID生成方式
            String nodeId = "node-" + inetAddr.getPort();
            recordActivity(nodeId);
        }
    }

    public void recordActivity(String nodeId) {
        lastHeardTimes.put(nodeId, System.currentTimeMillis());
        suspicionLevels.computeIfAbsent(nodeId, k -> new AtomicInteger(0)).set(0);
    }

    private void handleSuspectNode(String nodeId) {
        AtomicInteger suspicion = suspicionLevels.get(nodeId);
        if (suspicion == null) return;

        int level = suspicion.incrementAndGet();
        if (level == 1) {
            System.out.printf("[FD] Node %s is suspected (level %d)%n", nodeId, level);
        }

        // 达到一定怀疑级别后启动间接检查
        if (level >= 2) {
            indirectCheck(nodeId);
        }

        // 超过阈值则移除节点
        if (level > 5) {
            System.out.printf("[FD] Node %s is confirmed down%n", nodeId);
            gossipCore.getMembershipTable().getMembers().remove(nodeId);
            lastHeardTimes.remove(nodeId);
            suspicionLevels.remove(nodeId);
        }
    }

    private void indirectCheck(String suspectNodeId) {
        List<MemberState> targets = gossipCore.getMembershipTable()
                .selectRandomTargets(INDIRECT_CHECK_COUNT)
                .stream()
                .filter(m -> !m.getNodeId().equals(suspectNodeId))
                .toList();

        for (MemberState target : targets) {
            String probeMsg = "PROBE:" + suspectNodeId + ":" + gossipCore.getLocalNodeId();
            gossipCore.getTransport().send(target.getAddress(), probeMsg);
        }
    }

}
